/*-------------------------------------------------------------------------
 *
 * instrument.c
 *	 用于计划执行的仪表功能
 *
 *
 * Copyright (c) 2001-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/executor/instrument.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>

#include "executor/instrument.h"

BufferUsage pgBufferUsage;
static BufferUsage save_pgBufferUsage;
WalUsage	pgWalUsage;
static WalUsage save_pgWalUsage;

static void fc_BufferUsageAdd(BufferUsage *fc_dst, const BufferUsage *fc_add);
static void fc_WalUsageAdd(WalUsage *fc_dst, WalUsage *fc_add);


/* 分配新的仪器结构 */
Instrumentation *
InstrAlloc(int fc_n, int fc_instrument_options, bool fc_async_mode)
{
	Instrumentation *fc_instr;

	/* 将所有字段初始化为零，然后根据需要修改 */
	fc_instr = palloc0(fc_n * sizeof(Instrumentation));
	if (fc_instrument_options & (INSTRUMENT_BUFFERS | INSTRUMENT_TIMER | INSTRUMENT_WAL))
	{
		bool		fc_need_buffers = (fc_instrument_options & INSTRUMENT_BUFFERS) != 0;
		bool		fc_need_wal = (fc_instrument_options & INSTRUMENT_WAL) != 0;
		bool		fc_need_timer = (fc_instrument_options & INSTRUMENT_TIMER) != 0;
		int			fc_i;

		for (fc_i = 0; fc_i < fc_n; fc_i++)
		{
			fc_instr[fc_i].need_bufusage = fc_need_buffers;
			fc_instr[fc_i].need_walusage = fc_need_wal;
			fc_instr[fc_i].need_timer = fc_need_timer;
			fc_instr[fc_i].async_mode = fc_async_mode;
		}
	}

	return fc_instr;
}

/* 初始化一个预分配的仪器结构。 */
void InstrInit(Instrumentation *fc_instr, int fc_instrument_options)
{
	memset(fc_instr, 0, sizeof(Instrumentation));
	fc_instr->need_bufusage = (fc_instrument_options & INSTRUMENT_BUFFERS) != 0;
	fc_instr->need_walusage = (fc_instrument_options & INSTRUMENT_WAL) != 0;
	fc_instr->need_timer = (fc_instrument_options & INSTRUMENT_TIMER) != 0;
}

/* 进入一个计划节点 */
void InstrStartNode(Instrumentation *fc_instr)
{
	if (fc_instr->need_timer &&
		!INSTR_TIME_SET_CURRENT_LAZY(fc_instr->starttime))
		elog(ERROR, "InstrStartNode called twice in a row");

	/* 如果需要，在节点入口处保存缓冲区使用总数 */
	if (fc_instr->need_bufusage)
		fc_instr->bufusage_start = pgBufferUsage;

	if (fc_instr->need_walusage)
		fc_instr->walusage_start = pgWalUsage;
}

/* 从一个计划节点退出 */
void InstrStopNode(Instrumentation *fc_instr, double fc_nTuples)
{
	double		fc_save_tuplecount = fc_instr->tuplecount;
	instr_time	fc_endtime;

	/* 计数返回的元组 */
	fc_instr->tuplecount += fc_nTuples;

	/* 只有在请求计时器的情况下才更新时间 */
	if (fc_instr->need_timer)
	{
		if (INSTR_TIME_IS_ZERO(fc_instr->starttime))
			elog(ERROR, "InstrStopNode called without start");

		INSTR_TIME_SET_CURRENT(fc_endtime);
		INSTR_TIME_ACCUM_DIFF(fc_instr->counter, fc_endtime, fc_instr->starttime);

		INSTR_TIME_SET_ZERO(fc_instr->starttime);
	}

	/* 将进入节点以来的缓冲区使用增量加到总数中 */
	if (fc_instr->need_bufusage)
		BufferUsageAccumDiff(&fc_instr->bufusage,
							 &pgBufferUsage, &fc_instr->bufusage_start);

	if (fc_instr->need_walusage)
		WalUsageAccumDiff(&fc_instr->walusage,
						  &pgWalUsage, &fc_instr->walusage_start);

	/* 这是本周期的第一个元组吗？ */
	if (!fc_instr->running)
	{
		fc_instr->running = true;
		fc_instr->firsttuple = INSTR_TIME_GET_DOUBLE(fc_instr->counter);
	}
	else
	{
		/*
		 * 在异步模式下，如果计划节点之前没有发出任何元组，
		 * 这可能是第一个元组
 */
		if (fc_instr->async_mode && fc_save_tuplecount < 1.0)
			fc_instr->firsttuple = INSTR_TIME_GET_DOUBLE(fc_instr->counter);
	}
}

/* 更新元组计数 */
void InstrUpdateTupleCount(Instrumentation *fc_instr, double fc_nTuples)
{
	/* 计数返回的元组 */
	fc_instr->tuplecount += fc_nTuples;
}

/* 完成计划节点的运行周期 */
void InstrEndLoop(Instrumentation *fc_instr)
{
	double		fc_totaltime;

	/* 如果没有发生任何事情，或者已经关闭，则跳过 */
	if (!fc_instr->running)
		return;

	if (!INSTR_TIME_IS_ZERO(fc_instr->starttime))
		elog(ERROR, "InstrEndLoop called on running node");

	/* 将每周期的统计信息累加到总数中 */
	fc_totaltime = INSTR_TIME_GET_DOUBLE(fc_instr->counter);

	fc_instr->startup += fc_instr->firsttuple;
	fc_instr->total += fc_totaltime;
	fc_instr->ntuples += fc_instr->tuplecount;
	fc_instr->nloops += 1;

	/* 为下一个周期重置（如果有） */
	fc_instr->running = false;
	INSTR_TIME_SET_ZERO(fc_instr->starttime);
	INSTR_TIME_SET_ZERO(fc_instr->counter);
	fc_instr->firsttuple = 0;
	fc_instr->tuplecount = 0;
}

/* 汇总仪器信息 */
void InstrAggNode(Instrumentation *fc_dst, Instrumentation *fc_add)
{
	if (!fc_dst->running && fc_add->running)
	{
		fc_dst->running = true;
		fc_dst->firsttuple = fc_add->firsttuple;
	}
	else if (fc_dst->running && fc_add->running && fc_dst->firsttuple > fc_add->firsttuple)
		fc_dst->firsttuple = fc_add->firsttuple;

	INSTR_TIME_ADD(fc_dst->counter, fc_add->counter);

	fc_dst->tuplecount += fc_add->tuplecount;
	fc_dst->startup += fc_add->startup;
	fc_dst->total += fc_add->total;
	fc_dst->ntuples += fc_add->ntuples;
	fc_dst->ntuples2 += fc_add->ntuples2;
	fc_dst->nloops += fc_add->nloops;
	fc_dst->nfiltered1 += fc_add->nfiltered1;
	fc_dst->nfiltered2 += fc_add->nfiltered2;

	/* 将进入节点以来的缓冲区使用增量加到总数中 */
	if (fc_dst->need_bufusage)
		fc_BufferUsageAdd(&fc_dst->bufusage, &fc_add->bufusage);

	if (fc_dst->need_walusage)
		fc_WalUsageAdd(&fc_dst->walusage, &fc_add->walusage);
}

/* 在并行执行器启动期间注意当前值 */
void InstrStartParallelQuery(void)
{
	save_pgBufferUsage = pgBufferUsage;
	save_pgWalUsage = pgWalUsage;
}

/* 在并行执行器关闭后报告使用情况 */
void InstrEndParallelQuery(BufferUsage *fc_bufusage, WalUsage *fc_walusage)
{
	memset(fc_bufusage, 0, sizeof(BufferUsage));
	BufferUsageAccumDiff(fc_bufusage, &pgBufferUsage, &save_pgBufferUsage);
	memset(fc_walusage, 0, sizeof(WalUsage));
	WalUsageAccumDiff(fc_walusage, &pgWalUsage, &save_pgWalUsage);
}

/* 在领导者的统计信息中累积工作量 */
void InstrAccumParallelQuery(BufferUsage *fc_bufusage, WalUsage *fc_walusage)
{
	fc_BufferUsageAdd(&pgBufferUsage, fc_bufusage);
	fc_WalUsageAdd(&pgWalUsage, fc_walusage);
}

/* dst += add */
static void fc_BufferUsageAdd(BufferUsage *fc_dst, const BufferUsage *fc_add)
{
	fc_dst->shared_blks_hit += fc_add->shared_blks_hit;
	fc_dst->shared_blks_read += fc_add->shared_blks_read;
	fc_dst->shared_blks_dirtied += fc_add->shared_blks_dirtied;
	fc_dst->shared_blks_written += fc_add->shared_blks_written;
	fc_dst->local_blks_hit += fc_add->local_blks_hit;
	fc_dst->local_blks_read += fc_add->local_blks_read;
	fc_dst->local_blks_dirtied += fc_add->local_blks_dirtied;
	fc_dst->local_blks_written += fc_add->local_blks_written;
	fc_dst->temp_blks_read += fc_add->temp_blks_read;
	fc_dst->temp_blks_written += fc_add->temp_blks_written;
	INSTR_TIME_ADD(fc_dst->blk_read_time, fc_add->blk_read_time);
	INSTR_TIME_ADD(fc_dst->blk_write_time, fc_add->blk_write_time);
	INSTR_TIME_ADD(fc_dst->temp_blk_read_time, fc_add->temp_blk_read_time);
	INSTR_TIME_ADD(fc_dst->temp_blk_write_time, fc_add->temp_blk_write_time);
}

/* dst += add - sub */
void BufferUsageAccumDiff(BufferUsage *fc_dst,
					 const BufferUsage *fc_add,
					 const BufferUsage *fc_sub)
{
	fc_dst->shared_blks_hit += fc_add->shared_blks_hit - fc_sub->shared_blks_hit;
	fc_dst->shared_blks_read += fc_add->shared_blks_read - fc_sub->shared_blks_read;
	fc_dst->shared_blks_dirtied += fc_add->shared_blks_dirtied - fc_sub->shared_blks_dirtied;
	fc_dst->shared_blks_written += fc_add->shared_blks_written - fc_sub->shared_blks_written;
	fc_dst->local_blks_hit += fc_add->local_blks_hit - fc_sub->local_blks_hit;
	fc_dst->local_blks_read += fc_add->local_blks_read - fc_sub->local_blks_read;
	fc_dst->local_blks_dirtied += fc_add->local_blks_dirtied - fc_sub->local_blks_dirtied;
	fc_dst->local_blks_written += fc_add->local_blks_written - fc_sub->local_blks_written;
	fc_dst->temp_blks_read += fc_add->temp_blks_read - fc_sub->temp_blks_read;
	fc_dst->temp_blks_written += fc_add->temp_blks_written - fc_sub->temp_blks_written;
	INSTR_TIME_ACCUM_DIFF(fc_dst->blk_read_time,
						  fc_add->blk_read_time, fc_sub->blk_read_time);
	INSTR_TIME_ACCUM_DIFF(fc_dst->blk_write_time,
						  fc_add->blk_write_time, fc_sub->blk_write_time);
	INSTR_TIME_ACCUM_DIFF(fc_dst->temp_blk_read_time,
						  fc_add->temp_blk_read_time, fc_sub->temp_blk_read_time);
	INSTR_TIME_ACCUM_DIFF(fc_dst->temp_blk_write_time,
						  fc_add->temp_blk_write_time, fc_sub->temp_blk_write_time);
}

/* 用于WAL使用量累积的辅助函数 */
static void fc_WalUsageAdd(WalUsage *fc_dst, WalUsage *fc_add)
{
	fc_dst->wal_bytes += fc_add->wal_bytes;
	fc_dst->wal_records += fc_add->wal_records;
	fc_dst->wal_fpi += fc_add->wal_fpi;
}

void WalUsageAccumDiff(WalUsage *fc_dst, const WalUsage *fc_add, const WalUsage *fc_sub)
{
	fc_dst->wal_bytes += fc_add->wal_bytes - fc_sub->wal_bytes;
	fc_dst->wal_records += fc_add->wal_records - fc_sub->wal_records;
	fc_dst->wal_fpi += fc_add->wal_fpi - fc_sub->wal_fpi;
}
